package com.new1cloud.file.upload.client.bootstrap.handler;

import com.new1cloud.file.domain.*;
import com.new1cloud.file.tools.FileSlicerTool;
import com.new1cloud.file.upload.client.entry.ChunkFile;
import io.netty.channel.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import static com.new1cloud.file.upload.client.bootstrap.FileUploadClient.defaultChunkSize;
import static com.new1cloud.file.upload.client.bootstrap.FileUploadClient.mapChunkFileNameByClient;

public class FileUploadClientHandler extends ChannelInboundHandlerAdapter implements LocalCache {
    protected final Logger log = LoggerFactory.getLogger(this.getClass());
    private Channel channel;
    private final UploadProgressListener uploadProgressListener;
    //计数器
    private final CountDownLatch latch;

    public FileUploadClientHandler(UploadProgressListener uploadProgressListener, CountDownLatch latch) {
        this.uploadProgressListener = uploadProgressListener;
        this.latch = latch;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.channel = ctx.channel();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof ChunkFileUploadStatus) {
            log.info("客户端收到来自服务端的ChunkFileUploadStatus类型消息");
            ChunkFileUploadStatus chunkFileUploadStatus = (ChunkFileUploadStatus) msg;
            uploadProgressListener.progressChanged(FileUploadProgressStatus.builder().fileName(chunkFileUploadStatus.getFileName())
                    .transferPercentage(chunkFileUploadStatus.getUploadPercentage()).build());
            //表示上传失败，客户端进行重试上传该分片文件
            if(Objects.equals(0, chunkFileUploadStatus.getUploadStatus())) {
                log.warn("文件({})的分片文件({})上传失败,将重新上传.", chunkFileUploadStatus.getFileName(), chunkFileUploadStatus.getChunkName());
                FileChunkInfo chunkInfo = chunkFiles.get(chunkFileUploadStatus.getFileHash());
                if(Objects.isNull(chunkInfo)) {
                    log.error("未查询到文件({})的分片文件({})的相关信息，将停止上传该文件.", chunkFileUploadStatus.getFileName(), chunkFileUploadStatus.getChunkName());
                    return;
                }
                chunkInfo.getChunkFiles().stream()
                        .filter(cf -> Objects.equals(chunkFileUploadStatus.getChunkName(), cf.getChunkName()))
                        .findAny()
                        .ifPresent(chunkFile -> ctx.writeAndFlush(FileSlicerTool.gotChunkFileBytes(ctx,chunkInfo.getFilePath(),
                                chunkFile.getOffset(),
                                defaultChunkSize,
                                chunkFile)
                        ).addListener(future -> {
//                            boolean re = chunkFile.getChunkData().release();
                            chunkFile.setChunkData(null);
                            log.info("文件({})的分片文件({})已上传，内存释放: {}", chunkFileUploadStatus.getFileName(), chunkFile.getChunkName(), true);
                        }));
                return;
            }
            //上传成功，则检查确认当前客户端是否已上传完其负责上传的所有分片文件
            log.debug("文件({})的分片文件({})已上传成功.", chunkFileUploadStatus.getFileName(), chunkFileUploadStatus.getChunkName());
            AtomicBoolean isCompleted = new AtomicBoolean(true);
            Set<String> completedUploadChunkFileNames = chunkFileUploadStatus.getFileUploadProcess().getCompletedChunkItem().stream().map(String::valueOf).collect(Collectors.toSet());
            Set<String> needUploadChunkFileNames = mapChunkFileNameByClient.get(channel);
            needUploadChunkFileNames.stream()
                    .filter(fileName -> !completedUploadChunkFileNames.contains(String.valueOf(fileName)))
                    .findAny()
                    .ifPresent(fileName -> isCompleted.compareAndSet(true, false));
            if(isCompleted.get()) {
                log.info("当前客户端已上传所有的分片文件，将关闭该客户端连接.");
                sendUploadCompleteMarkInfo(ctx, chunkFileUploadStatus.getFileUploadProcess(), latch);
                closeConnection(channel, latch);
            }
        }else if(msg instanceof SendChunkFilesReq) {
            SendChunkFilesReq sendChunkFilesReq = (SendChunkFilesReq) msg;
            //获取到服务端返回的分片文件上传的进度信息
            final FileUploadProcess fileUploadProcess = sendChunkFilesReq.getFileUploadProcess();
            final FileChunkInfo needSendChunkFiles = sendChunkFilesReq.getNeedSendChunkFiles();
            //表示服务端未查询到文件上传进度信息
            if(Objects.equals(-1L, fileUploadProcess.getFileSize())) {
                log.error("服务端未查询到文件({})的上传进度信息", fileUploadProcess.getFileName());
                uploadProgressListener.progressChanged(FileUploadProgressStatus.builder().fileName(fileUploadProcess.getFileName())
                        .transferPercentage(0).build());
                return;
            }

            //表示该文件已上传完成
            if(Objects.equals(1, fileUploadProcess.getUploadStatus())) {
                uploadProgressListener.progressChanged(FileUploadProgressStatus.builder().fileName(fileUploadProcess.getFileName())
                        .transferPercentage(100).build());
                log.info("文件({})已成功上传.", fileUploadProcess.getFileName());
                sendUploadCompleteMarkInfo(ctx, fileUploadProcess, latch);
                closeConnection(channel, latch);
                return;
            }

            //表示该文件未上传完成，则客户端开始进行文件切分上传ChunkFile
            FileChunkInfo fileChunkInfo = chunkFiles.get(fileUploadProcess.getFileHash());
            if(Objects.isNull(fileChunkInfo)) {
                log.error("未查询到文件({})的相关信息，将停止上传该文件.", fileUploadProcess.getFileName());
                return;
            }

            final double totalChunkNum = fileChunkInfo.getChunkFiles().size() * 1.0;
            int completedChunkItemCount = 0;
            //初始化本次待上传的分片文件列表
            List<ChunkFile> uncompletedChunkItem = needSendChunkFiles.getChunkFiles();
            //本地缓存当前客户端负责上传的分片文件名列表
            uncompletedChunkItem.forEach(chunkFile -> {
                Set<String> fileNames = mapChunkFileNameByClient.computeIfAbsent(channel, key -> new HashSet<>());
                fileNames.add(chunkFile.getChunkName());
            });
            //查看已上传到服务端的分片文件名
            if(!CollectionUtils.isEmpty(fileUploadProcess.getCompletedChunkItem())) {
                log.warn("检测到之前已上传该文件({})的部分分片文件到服务端.", fileUploadProcess.getFileName());
                final Set<Integer> completedChunkItem = new HashSet<>(fileUploadProcess.getCompletedChunkItem());
                //将已上传的分片文件从needSendChunkFiles里去除掉
                uncompletedChunkItem = needSendChunkFiles.getChunkFiles().stream()
                        .filter(chunkFile -> !completedChunkItem.contains(Integer.parseInt(chunkFile.getChunkName())))
                        .collect(Collectors.toList());
                completedChunkItemCount = completedChunkItem.size();
            }
            //首次展示上传进度信息
            uploadProgressListener.progressChanged(FileUploadProgressStatus.builder().fileName(fileUploadProcess.getFileName())
                    .transferPercentage((int) ((completedChunkItemCount / totalChunkNum) * 100)).build());
            //如果uncompletedChunkItem为空，表示已上传完所有的分片文件，直接关闭客户端连接
            if(CollectionUtils.isEmpty(uncompletedChunkItem)) {
                log.info("当前客户端已上传所有的分片文件，将关闭连接.");
                sendUploadCompleteMarkInfo(ctx, fileUploadProcess, latch);
                closeConnection(channel, latch);
                return;
            }
            //切分文件并依次发送分片文件到服务端
            uncompletedChunkItem.stream().parallel().forEach(chunkFile -> {
                log.info("将开始发送文件({})({})分片:{}.", fileUploadProcess.getFileName(), fileUploadProcess.getFileHash(), chunkFile.getChunkName());
                ctx.writeAndFlush(FileSlicerTool.gotChunkFileBytes(ctx,fileChunkInfo.getFilePath(),
                                chunkFile.getOffset(),
                                defaultChunkSize,
                                chunkFile)
                        )
                        .addListener(future -> {
//                    boolean re = chunkFile.getChunkData().release();
                            if(future.isSuccess()) {
                                //等待gc去回收
                                chunkFile.setChunkData(null);
                                log.debug("文件({})的分片文件({})已上传，内存释放: {}", fileUploadProcess.getFileName(), chunkFile.getChunkName(), true);
                            } else {
                                log.error("文件({})的分片文件({})上传异常，", fileUploadProcess.getFileName(), chunkFile.getChunkName(), future.cause());
                            }
                        });
//                try {
//                    Thread.sleep(500);
//                } catch (InterruptedException e) {
//                    throw new RuntimeException(e);
//                }
            });
        } else {
            log.error("客户端[FileUploadClientHandler]收到非预期类型的消息:{},将被丢弃!", msg);
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        log.error("客户端发生异常: ", cause);
        ctx.close();
    }

    /**
     * 关闭channel
     * @param channel channel
     */
    private void closeConnection(Channel channel, CountDownLatch latch) {
        if (channel != null) {
            // 添加一个监听器，确保关闭操作完成后执行
            channel.close().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) {
                    if (future.isSuccess()) {
                        log.info("NettyClient Connection closed successfully.");
                        latch.countDown();
                    } else {
                        log.error("NettyClient Connection closed failed.");
                    }
                }
            });
        }
    }

    /**
     * 通知服务端指定的文件已全部上传完毕
     */
    private void sendUploadCompleteMarkInfo(ChannelHandlerContext ctx, FileUploadProcess fileUploadProcess, CountDownLatch latch){
        if(Objects.equals(latch.getCount(), 1L)) {
            uploadProgressListener.progressChanged(FileUploadProgressStatus.builder().fileName(fileUploadProcess.getFileName())
                    .transferPercentage(100).build());
            ctx.writeAndFlush(UploadCompleteMark.builder().fileSize(fileUploadProcess.getFileSize())
                    .fileName(fileUploadProcess.getFileName()).fileHash(fileUploadProcess.getFileHash())
                            .uploadPath(fileUploadProcess.getUploadPath()).build()
                    )
                    .addListener(future -> {
                        if(future.isSuccess()) {
                            log.info("文件({})已全部上传至服务端，并完成通知.", fileUploadProcess.getFileName());
                        }
                    });
        }
    }

    /**
     * 发送待上传文件的基本信息(以获取该文件的分片文件的上传进度信息)
     * @param basicInfo 待上传文件的基本信息
     */
    public void sendUploadFileInfo(FileUploadBasicInfo basicInfo) {
        if(Objects.nonNull(basicInfo)) {
            this.channel.writeAndFlush(basicInfo).addListener(future -> {
                log.info("文件({})已向服务端发送成功.", basicInfo.getFileName());
            });
        }
    }
}
